package com.yangchao.webflux.projectreactor;

import org.junit.jupiter.api.DisplayName;
import org.junit.jupiter.api.Test;
import org.reactivestreams.Subscriber;
import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

import java.io.IOException;
import java.time.Duration;
import java.util.function.Consumer;

public class FluxTest {

    static Logger logger=org.slf4j.LoggerFactory.getLogger(FluxTest.class);
    @Test
    @DisplayName("Flux测试")
    public void testFlux01() {
        Flux<String> publisher = Flux.just("foo", "bar", "foobar");
        publisher.subscribe(new Consumer<String>() {
            @Override
            public void accept(String s) {
                logger.info("accept {}", s);
            }
        });
        publisher.subscribe(new Subscriber<String>() {
            private Subscription subscription;
            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                this.subscription.request(1);
            }

            @Override
            public void onNext(String s) {
                logger.info("subscriber2 {}", s);
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                logger.info("subscriber2 error {}", throwable.getMessage());
            }

            @Override
            public void onComplete() {
                logger.info("subscriber2 complete");
            }
        });
    }


    @Test
    @DisplayName("Flux测试2")
    public void testFlux02() throws IOException {
        Flux<Long> publisher = Flux.interval(Duration.ofSeconds(1));
        publisher.subscribe(new Consumer<Long>() {
            @Override
            public void accept(Long aLong) {
                logger.info("accept {}", aLong);
            }
        });
        System.in.read();
    }

    @Test
    @DisplayName("Flux测试3 doxapi")
    public void testFlux03() {
        Flux<String> publisher = Flux.just("foo", "bar", "foobar")
                .doOnComplete(() -> logger.info("doOnComplete"))
                .doOnTerminate(() -> logger.info("doOnTerminate"))
                .doFinally(signalType -> {
                            if (signalType == SignalType.ON_COMPLETE) {
                                logger.info("Flux processing completed.");
                            } else if (signalType == SignalType.ON_ERROR) {
                                logger.info("Flux processing encountered an error.");
                            } else {
                                logger.info("Flux processing terminated.");
                            }
                        }
                ).doFirst(() -> logger.info("doFirst"))
                        .doOnRequest(n -> logger.info("n = " + n));


        publisher.subscribe(
                v -> {
                    // if ("foo".equals(v)) {
                    //     throw new RuntimeException("foo");
                    // }
                    logger.info("v = " + v);

                }, // 流元素消费
                throwable -> logger.info("throwable = " + throwable), // 感知异常结束
                () -> logger.info("流结束了...") // 感知正常结束
        );

        // publisher.subscribe(
        //         v -> {
        //             // if ("foo".equals(v)) {
        //             //     throw new RuntimeException("foo");
        //             // }
        //             logger.info("v2 = " + v);
        //
        //         }
        // );
    }

    @Test
    @DisplayName("Flux测试3 doxapi doOnRequest")
    public void testFlux04() {
        Flux<String> publisher = Flux.just("foo", "bar", "foobar")
                .log()
                .doOnComplete(() -> logger.info("doOnComplete"))
                .doOnTerminate(() -> logger.info("doOnTerminate"))
                .doFinally(signalType -> {
                            if (signalType == SignalType.ON_COMPLETE) {
                                logger.info("Flux processing completed.");
                            } else if (signalType == SignalType.ON_ERROR) {
                                logger.info("Flux processing encountered an error.");
                            } else {
                                logger.info("Flux processing terminated.");
                            }
                        }
                ).doFirst(() -> logger.info("doFirst"))
                .doOnRequest(n -> logger.info("doOnRequest n = " + n))
                        .doOnCancel(() -> logger.info("doOnCancel"))
                .doOnNext(v -> logger.info(" doOnNext v = " + v));


        publisher.subscribe(new Subscriber<String>() {
            private Subscription subscription;
            @Override
            public void onSubscribe(Subscription subscription) {
                this.subscription = subscription;
                this.subscription.request(1);
            }

            @Override
            public void onNext(String s) {
                logger.info("subscriber2 {}", s);
                this.subscription.request(1);
            }

            @Override
            public void onError(Throwable throwable) {
                logger.info("subscriber2 error {}", throwable.getMessage());
            }

            @Override
            public void onComplete() {
                logger.info("subscriber2 complete");
            }
        });
    }


}
